package org.future.task.core.handler;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

public class DefaultConnectionImpl implements Connection {

    private final ConcurrentHashMap<String, LinkedBlockingQueue<Object>> concurrentMap = new ConcurrentHashMap<>();

    private int capacity = 1000;

    public DefaultConnectionImpl(int capacity) {
        if (capacity > 0)
            this.capacity = capacity;
    }

    @Override
    public List<Object> getDataList(String queueName, int size, int timeOut) {
        LinkedBlockingQueue<Object> orDefault = concurrentMap.computeIfAbsent(queueName, (LinkedBlockingQueue) -> new LinkedBlockingQueue<>(capacity));
        List<Object> dataList = new ArrayList<>();
        try {
            Object take = orDefault.take();
            dataList.add(take);
            orDefault.drainTo(dataList, size - 1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return dataList;
    }

    @Override
    public long size(String queueName) {
        LinkedBlockingQueue<Object> orDefault = concurrentMap.computeIfAbsent(queueName, (linkedBlockingQueue) -> new LinkedBlockingQueue<>(capacity));
        return orDefault.size();
    }

    @Override
    public void sendDataList(String queueName, List<?> dataList) {
        for (Object obj : dataList) {
            sendData(queueName, obj);
        }
    }

    @Override
    public void sendData(String queueName, Object data) {
        LinkedBlockingQueue<Object> orDefault = concurrentMap.computeIfAbsent(queueName, (linkedBlockingQueue) -> new LinkedBlockingQueue<>(capacity));
        try {
            orDefault.put(data);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String getType() {
        return LinkedBlockingQueue.class.getSimpleName();
    }

    @Override
    public Commit getCommit() {
        return null;
    }


}
